package com.pudaocredit.function;

import com.pudaocredit.Exception.NoTableException;
import com.pudaocredit.constans.HTableConst;
import com.pudaocredit.serializer.QueryHbaseCaihongTblUdfSerializer;
import com.pudaocredit.utils.HbaseConUtil;
import org.apache.hadoop.hbase.ClusterStatus;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hive.com.esotericsoftware.kryo.DefaultSerializer;
import org.slf4j.LoggerFactory;

import java.io.IOException;
@DefaultSerializer(value = QueryHbaseCaihongTblUdfSerializer.class)
public class QueryHbaseCaihongTblUdf extends GenericUDF {
    private static org.slf4j.Logger logger = LoggerFactory.getLogger(QueryHbaseCaihongTblUdf.class);
    Connection connection = null;
    Admin admin = null;
    String colfamily;
//    private static final String KEYTAB_FILE =QueryHbaseCaihongTblUdf.class.getClassLoader().getResource("kerb/hive.keytab").getPath();

//    private static final String KERB_CONF =QueryHbaseCaihongTblUdf.class.getClassLoader().getResource("kerb/krb5.conf").getPath();

    @Override
    public ObjectInspector initialize(ObjectInspector[] arguments) throws UDFArgumentException {
    /*try{
        connection = HbaseConUtil.getConnection("11.243.32.178,11.243.32.179,11.243.32.177", "2181", "/export/kerberos/1/hive.keytab", "/etc/krb5.conf", "hive/ocean@PUDAO.COM");
        admin = connection.getAdmin();
        table = connection.getTable(TableName.valueOf(HTableConst.MOBILE_SHA256_TABLE));
        if (admin.tableExists(TableName.valueOf(HTableConst.MOBILE_SHA256_TABLE))){
            System.out.println(HTableConst.MOBILE_SHA256_TABLE+"存在");

        }
//        connection = HbaseConUtil.getConnection("11.243.32.178,11.243.32.179,11.243.32.177", "2181", KEYTAB_FILE, KERB_CONF, "hive/ocean@PUDAO.COM");
    }catch (Exception e){
        System.out.println("创建hbase连接失败");
        logger.info("创建hbase连接失败");
        e.printStackTrace();
    }*/

        if (arguments.length != 3) {
            throw new UDFArgumentException("Input args length  erro!!!");
        } else if (!arguments[0].getCategory().equals(ObjectInspector.Category.PRIMITIVE)) {
            throw new UDFArgumentTypeException(0, "Input args  type error!!");
        } else {
            return PrimitiveObjectInspectorFactory.javaStringObjectInspector;
        }
    }

    @Override
    public String evaluate(DeferredObject[] arguments) throws HiveException {
        String str = arguments[0].get().toString();
        String info_type = arguments[1].get().toString();
        String hash_type = arguments[2].get().toString();
        System.out.println("参数str："+str+",信息类型为："+info_type+"，hash类型为"+hash_type);
        Table table = null;

        if (table == null){
            try{
                System.out.println("初始化hbase客户端.......");
                logger.info("初始化hbase客户端.......");
                connection = HbaseConUtil.getConnection("11.243.32.178,11.243.32.179,11.243.32.177", "2181", "/export/kerberos/1/hbase.keytab", "/etc/krb5.conf", "hbase/ocean@PUDAO.COM");

                if (HTableConst.INFO_TYPE_MOBILE.equals(info_type) && HTableConst.HASH_MD5.equals(hash_type)){
                    table = connection.getTable(TableName.valueOf(HTableConst.MOBILE_MD5_TABLE));
                    colfamily=HTableConst.COL_FAMILY_MOBILE;
                }else if(HTableConst.INFO_TYPE_MOBILE.equals(info_type) && HTableConst.HASH_SHA256.equals(hash_type)){
                    table = connection.getTable(TableName.valueOf(HTableConst.MOBILE_SHA256_TABLE));
                    colfamily=HTableConst.COL_FAMILY_MOBILE;
                }else if(HTableConst.INFO_TYPE_MOBILE.equals(info_type) && HTableConst.HASH_SM3.equals(hash_type)){
                    table = connection.getTable(TableName.valueOf(HTableConst.MOBILE_SM3_TABLE));
                    colfamily=HTableConst.COL_FAMILY_MOBILE;
                }else if(HTableConst.INFO_TYPE_CERT.equals(info_type) && HTableConst.HASH_MD5.equals(hash_type)){
                    table = connection.getTable(TableName.valueOf(HTableConst.CERT_MD5_TABLE));
                    colfamily=HTableConst.COL_FAMILY_CERT;
                }else if(HTableConst.INFO_TYPE_CERT.equals(info_type) && HTableConst.HASH_SHA256.equals(hash_type)){
                    table = connection.getTable(TableName.valueOf(HTableConst.CERT_SHA256_TABLE));
                    colfamily=HTableConst.COL_FAMILY_CERT;
                }else if(HTableConst.INFO_TYPE_CERT.equals(info_type) && HTableConst.HASH_SM3.equals(hash_type)){
                    table = connection.getTable(TableName.valueOf(HTableConst.CERT_SM3_TABLE));
                    colfamily=HTableConst.COL_FAMILY_CERT;
                }else {
                    throw new NoTableException("无对应表信息，请核对传递参数是否正确");
                }
                System.out.println("查询hbase表为："+table+",列簇为："+colfamily);
            }catch (Exception e){
                e.printStackTrace();
            }
        }
        try{
            System.out.println("开始取数-->str:"+str+"，colfamily："+colfamily);
            byte[] row = new byte[]{};
            row = str.getBytes();
            System.out.println("step1");
            Get get = new Get(row);
            System.out.println("step2:"+table.getName().getNameAsString());
            Result result = table.get(get);
            System.out.println("step3"+result.toString());
            String valueStr = new String(result.getValue(Bytes.toBytes(colfamily),Bytes.toBytes(HTableConst.CELL)),"utf-8");
            System.out.println("查询结果为："+valueStr);
            return valueStr;
        }catch (Exception e){
            System.out.println("捕获异常");
            e.printStackTrace();
        }/*finally {
            try {
                if (connection != null) connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }*/
        return null;
    }

    @Override
    public String getDisplayString(String[] children) {
        return null;
    }

    @Override
    public void close() throws IOException {
        if (admin != null){
            admin.close();
        }
        if (connection != null){
            connection.close();
        }
//        super.close();
    }


}
